package com.hadoop.demo.hdfs;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

import java.io.BufferedReader;
import java.io.File;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.ArrayList;
import java.util.List;

public class HDFSUtil {

    private static final String ROOT_PATH = "/";

    /**
     * 列出对应hdfs下面的文件名
     *
     * @param configuration 配置文件
     * @param dir           路径
     * @return
     * @throws IOException
     * @throws URISyntaxException
     */
    public static List<String> listFiles(Configuration configuration, String dir) throws IOException, URISyntaxException {
        try {
            List<String> files = new ArrayList<>();
            FileSystem fs = FileSystem.get(new URI(dir), configuration);
            FileStatus[] fileStatuses = fs.listStatus(new Path(dir));
            for (FileStatus fileStatus : fileStatuses) {
                Path path = fileStatus.getPath();
                files.add(path.getName());
            }
            return files;
        } catch (Exception e) {
            throw e;
        }
    }

    /**
     * 创建目录
     *
     * @param configuration 配置
     * @param dir           路径
     * @throws IOException
     * @throws URISyntaxException
     */
    public static void createDir(Configuration configuration, String dir) throws IOException, URISyntaxException {
        FileSystem fs = FileSystem.get(new URI(ROOT_PATH), configuration);
        if (fs.exists(new Path(dir))) {
            throw new IOException(String.format("file path %s exist", dir));
        }
        fs.mkdirs(new Path(dir));
    }

    /**
     * 创建文件
     *
     * @param configuration 配置
     * @param fileName      文件名
     * @throws IOException
     * @throws URISyntaxException
     */
    public static void createFile(Configuration configuration, String fileName) throws IOException, URISyntaxException {
        FileSystem fs = FileSystem.get(new URI(ROOT_PATH), configuration);
        FSDataOutputStream output = fs.create(new Path(fileName));
        output.close();
    }

    /**
     * 向文件中写入数据
     *
     * @param configuration 配置
     * @param fileName      文件名
     * @param fileContent   文件内容
     * @throws IOException
     * @throws URISyntaxException
     */
    public static void writeFile(Configuration configuration, String fileName, String fileContent) throws IOException, URISyntaxException {
        byte[] bytes = fileContent.getBytes();
        FileSystem fs = FileSystem.get(new URI(ROOT_PATH), configuration);
        FSDataOutputStream output = fs.create(new Path(fileName));
        output.write(bytes);
        output.close();
    }

    /**
     * 删除文件
     *
     * @param configuration 配置
     * @param fileName      文件名或者文件目录
     * @throws IOException
     * @throws URISyntaxException
     */
    public static void deleteFile(Configuration configuration, String fileName) throws IOException, URISyntaxException {
        Path file = new Path(fileName);
        FileSystem fs = FileSystem.get(new URI(ROOT_PATH), configuration);
        if (fs.exists(file)) { // if exists, delete
            fs.delete(file, true);
        } else {
            throw new IOException(String.format("file name %s is not exist!", fileName));
        }
    }

    /**
     * 上传本地文件远程目录
     *
     * @param configuration 配置
     * @param localFile     本地文件
     * @param remoteDir     远程文件目录
     * @throws IOException
     * @throws URISyntaxException
     */
    public static void uploadFile(Configuration configuration, String localFile, String remoteDir) throws IOException, URISyntaxException {
        Path local = new Path(localFile);
        Path remote = new Path(remoteDir);
        if (!new File(localFile).exists()) {
            throw new IOException(String.format("local file name %s is not exist!", localFile));
        }

        FileSystem fs = FileSystem.get(new URI(ROOT_PATH), configuration);
        if (!fs.exists(remote)) {
            throw new IOException(String.format("remote file name %s is not exist!", remoteDir));
        }
        fs.copyFromLocalFile(local, remote);
    }

    /**
     * 将内容写道远程文件中
     *
     * @param configuration 配置
     * @param fileName      文件名
     * @param fileContent   文件内容
     * @throws IOException
     * @throws URISyntaxException
     */
    public static void appendFile(Configuration configuration, String fileName, String fileContent) throws IOException, URISyntaxException {
        FileSystem fs = FileSystem.get(new URI(ROOT_PATH), configuration);
        Path remote = new Path(fileName);
        byte[] bytes = fileContent.getBytes();
        if (!fs.exists(remote)) {
            createFile(configuration, fileName);
        }
        FSDataOutputStream remoteStream = fs.append(remote);
        remoteStream.write(bytes);
        remoteStream.close();
    }

    /**
     * 文件内容读取
     *
     * @param configuration 配置
     * @param fileName      文件名
     * @throws IOException
     */
    public static void readFile(Configuration configuration, String fileName) throws IOException {
        FileSystem fs = FileSystem.get(configuration);
        Path remotePath = new Path(fileName);
        FSDataInputStream in = fs.open(remotePath);
        BufferedReader d = new BufferedReader(new InputStreamReader(in));
        String line;
        while ((line = d.readLine()) != null) {
            System.out.println(line);
        }
        d.close();
        in.close();
        fs.close();
    }

    public static void main(String[] args) throws IOException, URISyntaxException {
        System.setProperty("HADOOP_USER_NAME", "hadoop");
        Configuration configuration = new Configuration();
        configuration.set("fs.defaultFS", "hdfs://172.16.1.229:8020");
        List<String> strings = listFiles(configuration, "/test/word-count");
        System.out.println(strings);
        deleteFile(configuration,"/test/word-count");
        readFile(configuration,"/test/word-count/part-r-00000");
        /*createDir(configuration, "/test");
        uploadFile(configuration,"D:\\tmp\\hadoop\\test.txt","/test/");*/
    }
}
